package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;

/**
 * 发送同步消息
 */
public class ASimpleTest {

    /**
     * 生产者
     * @throws Exception
     */
    @Test
    public void simpleProducer() throws Exception {
        // 创建一个生产者 (制定一个组名)
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
        // 连接namesrv
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        // 启动
        producer.start();
        // 循环发消息
        for (int i = 0; i < 10; i++) {
            // 创建一个消息
            Message message = new Message("testTopic", "hello rocketmq".getBytes());
            // 发送消息
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult.getSendStatus());
        }
        // 关闭
        producer.shutdown();
    }


    /**
     * 消费者
     * @throws Exception
     */
    @Test
    public void simpleConsumer() throws Exception {
        // 创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        // 消费者拉取消息模式
        // DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer("test-consumer-group");
        // 连接namesrv
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        // 订阅主题 * 代表订阅这个主题中的所有的消息 后期会有消息过滤
        consumer.subscribe("testTopic", "*");
        // 设置一个监听器 (一直监听 异步回调方式)
        consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
            // 这个就是一个消费的方法 (业务处理)
            System.out.println("我是消费者");
            System.out.println(list.get(0).toString());
            System.out.println("消息内容" + new String(list.get(0).getBody()));
            System.out.println("消息消费上下文" + context);
            // 返回值 CONSUME_SUCCESS成功, 消息会从mq中出队 如果返回
            // RECONSUME_LATER (报错/null) 失败 消息会重新回到mq队列 过一会重新投递出来 给当前消费者或者给其他消费者
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        // 启动
        consumer.start();
        // 挂起当前jvm
        System.in.read();
    }
}
